package com.fanxl.flink.datastream.transformation;

import com.fanxl.flink.datastream.kafka.WordPeriodicWatermark;
import com.fanxl.flink.datastream.source.AccessSource;
import org.apache.flink.api.common.functions.*;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * @description
 * @author: fanxl
 * @date: 2022/1/12 0012 0:01
 */
public class TransformationApp {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        map(env);
//        flatMap(env);
//        keyBy(env);
//        reduce(env);
        join(env);
        env.execute("TransformationApp");
    }

    public static void map(StreamExecutionEnvironment env) {
        DataStreamSource<String> dataStreamSource = env.readTextFile("data/access.data");
        dataStreamSource.map(new MapFunction<String, Access>() {
            @Override
            public Access map(String str) throws Exception {
                String[] arrays = str.split(",");
                Long time = Long.getLong(arrays[0].trim());
                Double traffic = Double.parseDouble(arrays[2]);
                return new Access(time, arrays[1], traffic);
            }
        })
        .filter(new FilterFunction<Access>() {
            @Override
            public boolean filter(Access access) throws Exception {
                return access.getTime() > 0;
            }
        });

        dataStreamSource.print();
    }

    public static void flatMap(StreamExecutionEnvironment env) {
        DataStreamSource<String> source = env.readTextFile("data/word.data");
        source.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> collector) throws Exception {
                String[] array = value.split(",");
                for (String str : array) {
                    collector.collect(str);
                }
            }
        }).print();
    }

    public static void keyBy(StreamExecutionEnvironment env) {
        DataStreamSource<String> dataStreamSource = env.readTextFile("data/access.data");
        SingleOutputStreamOperator<Access> accessStream = dataStreamSource.map(new MapFunction<String, Access>() {
            @Override
            public Access map(String str) throws Exception {
                String[] arrays = str.split(",");
                Long time = Long.parseLong(arrays[0].trim());
                Double traffic = Double.parseDouble(arrays[2]);
                return new Access(time, arrays[1], traffic);
            }
        });
        // 按照domain进行分组，再按照traffic进行求和
//        accessStream.keyBy("domain").sum("traffic").print();
        accessStream.keyBy(new KeySelector<Access, String>() {
            @Override
            public String getKey(Access access) throws Exception {
                return access.getDomain();
            }
        }).sum("traffic").print();
    }

    public static void reduce(StreamExecutionEnvironment env) {
        DataStreamSource<String> source = env.readTextFile("data/word.data");
        source.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> collector) throws Exception {
                String[] array = value.split(",");
                for (String str : array) {
                    collector.collect(str);
                }
            }
        }).map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return new Tuple2<>(value, 1);
            }
        }).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> x) throws Exception {
                return x.f0;
            }
        }).reduce(new ReduceFunction<Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> o1, Tuple2<String, Integer> o2) throws Exception {
                return Tuple2.of(o1.f0, o1.f1 + o2.f1);
            }
        }).print();
//        accessStream.keyBy(Access::getDomain).reduce(new ReduceFunction<Access>() {
//            @Override
//            public Access reduce(Access o1, Access o2) throws Exception {
//                return new Access(o1.getTime(), o1.getDomain(), o1.getTraffic() + o2.getTraffic());
//            }
//        }).print();
    }

    /**
     * 合并多个相同类型的流
     * @param env
     */
    public static void union(StreamExecutionEnvironment env) {
        DataStreamSource<Access> stream1 = env.addSource(new AccessSource());
        DataStreamSource<Access> stream2 = env.addSource(new AccessSource());
        DataStream<Access> union = stream1.union(stream2);
        union.print();
    }

    /**
     * 合并2个不同类型的流
     * @param env
     */
    public static void connect(StreamExecutionEnvironment env) {
        DataStreamSource<Access> stream1 = env.addSource(new AccessSource());
        SingleOutputStreamOperator<String> stream2 = env.readTextFile("data/word.data");
        ConnectedStreams<Access, String> connect = stream1.connect(stream2);
        SingleOutputStreamOperator<String> map = connect.map(new CoMapFunction<Access, String, String>() {
            @Override
            public String map1(Access value) throws Exception {
                return value.getDomain();
            }

            @Override
            public String map2(String value) throws Exception {
                return value;
            }
        });
        map.print();
    }

    public static void join(StreamExecutionEnvironment env) {
        DataStreamSource<String> dataStreamSource = env.readTextFile("data/access.data");
        SingleOutputStreamOperator<Map<String, Object>> mapStream1 = dataStreamSource.flatMap(new FlatMapFunction<String, Map<String, Object>>() {
            @Override
            public void flatMap(String value, Collector<Map<String, Object>> out) throws Exception {
                Map<String, Object> map1 = new HashMap<>();
                map1.put("type", "日志类型" + System.currentTimeMillis());
                map1.put("domain", "t");
                map1.put("dist_ip", "10.1.1.2");
                map1.put("timeStamp", System.currentTimeMillis());
                out.collect(map1);

                Map<String, Object> map2 = new HashMap<>();
                map2.put("type", "日志类型" + System.currentTimeMillis());
                map2.put("domain", "t");
                map2.put("dist_ip", "10.1.1.3");
                map2.put("timeStamp", System.currentTimeMillis());
                out.collect(map2);
            }
        }).assignTimestampsAndWatermarks(new WordPeriodicWatermark<>());

        SingleOutputStreamOperator<Map<String, Object>> mapStream2 = dataStreamSource.flatMap(new FlatMapFunction<String, Map<String, Object>>() {
            @Override
            public void flatMap(String value, Collector<Map<String, Object>> out) throws Exception {
                Map<String, Object> map1 = new HashMap<>();
                map1.put("type", "日志类型" + System.currentTimeMillis());
                map1.put("domain", "t");
                map1.put("dist_ip", "10.1.1.2");
                map1.put("timeStamp", System.currentTimeMillis());
                out.collect(map1);

                Map<String, Object> map2 = new HashMap<>();
                map2.put("type", "日志类型" + System.currentTimeMillis());
                map2.put("domain", "t");
                map2.put("dist_ip", "10.1.1.5");
                map2.put("timeStamp", System.currentTimeMillis());
                out.collect(map2);
            }
        }).assignTimestampsAndWatermarks(new WordPeriodicWatermark<>());

        mapStream1.join(mapStream2).where(new NameKeySelector("dist_ip")).equalTo(new NameKeySelector("dist_ip"))
                .window(TumblingEventTimeWindows.of(Time.minutes(5))).apply(new JoinFunction<Map<String, Object>, Map<String, Object>, Map<String, Object>>() {
            @Override
            public Map<String, Object> join(Map<String, Object> first, Map<String, Object> second) throws Exception {
                Map<String, Object> joinMap = new HashMap<>();
                joinMap.putAll(first);
                joinMap.putAll(second);
                return joinMap;
            }
        }).print();
    }


}
